package net.bw.realtime.jtp.dws.log.function;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import net.bw.realtime.jtp.common.utils.DateTimeUtil;
import net.bw.realtime.jtp.dws.log.bean.PageViewBean;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import java.util.HashMap;

/**
 * @author liuyawei
 * @date 2025/5/19
 */
public class PageViewBeanMapFunction extends RichMapFunction<String, PageViewBean> {

    // 1.声明状态
    private transient ValueState<String> lastVisitDateState = null;

    // 2.定义地理字典
    private HashMap<String, String> areaDic = null;

    @Override
    public void open(Configuration parameters) throws Exception {

        // 2.初始化状态
        ValueStateDescriptor<String> valueStateDescriptor = new ValueStateDescriptor<>(
                "lastVisitDateState", String.class);

        // 设置生命周期q
        StateTtlConfig ttlConfig = StateTtlConfig
                .newBuilder(Time.days(1))
                .updateTtlOnReadAndWrite()
                .neverReturnExpired()
                .build();

        valueStateDescriptor.enableTimeToLive(ttlConfig);

        lastVisitDateState = getRuntimeContext().getState(valueStateDescriptor);

        // 3.初始化字典
        areaDic = getAreaDic();

    }

    @Override
    public PageViewBean map(String value) throws Exception {

        //  1.获取数据
        JSONObject jsonObject = JSON.parseObject(value);
        JSONObject common = jsonObject.getJSONObject("common");
        JSONObject page = jsonObject.getJSONObject("page");

        // 根据地理字典获取省份
        String arValue = common.getString("ar");
        String province = areaDic.get(arValue);

        // 访客数
        Long uvCount = 0L;

        // 2.获取上一次时间
        String lastVisitDate = lastVisitDateState.value();

        // 3.获取当前时间
        String currentDate = DateTimeUtil.convertLongToString(jsonObject.getLong("ts"), DateTimeUtil.DATE_FORMAT);

        // 4.判断是否是第一次访问
        // 值为null，则表示第一次访问
        // 值不为null，则判断时间，则判断是否是同一天，则不是第一次访问，否则是第一次访问
        if(null == lastVisitDate || !lastVisitDate.equals(currentDate)){
            // 更新状态
            uvCount = 1L;
            lastVisitDateState.update(currentDate);
        }

        // 5.创建实例
        PageViewBean pageViewBean = new PageViewBean(
                        null ,
                        null ,
                        province ,
                        common.getString("ba") ,
                        common.getString("ch") ,
                        common.getString("is_new") ,
                        1L ,
                        page.getLong("during_time") ,
                        uvCount,
                        page.get("last_page_id") == null ? 1L : 0L ,
                        jsonObject.getLong("ts")
        );

        return pageViewBean;
    }

    // 获取字典
    private HashMap<String, String> getAreaDic() {
        HashMap<String, String> map = new HashMap<>();
        map.put("110000","北京");
        map.put("120000","天津");
        map.put("140000","山西");
        map.put("150000","内蒙古");
        map.put("130000","河北");
        map.put("310000","上海");
        map.put("320000","江苏");
        map.put("330000","浙江");
        map.put("340000","安徽");
        map.put("350000","福建");
        map.put("360000","江西");
        map.put("370000","山东");
        map.put("710000","台湾");
        map.put("230000","黑龙江");
        map.put("220000","吉林");
        map.put("210000","辽宁");
        map.put("610000","陕西");
        map.put("620000","甘肃");
        map.put("630000","青海");
        map.put("640000","宁夏");
        map.put("650000","新疆");
        map.put("410000","河南");
        map.put("420000","湖北");
        map.put("430000","湖南");
        map.put("440000","广东");
        map.put("450000","广西");
        map.put("460000","海南");
        map.put("810000","香港");
        map.put("820000","澳门");
        map.put("510000","四川");
        map.put("520000","贵州");
        map.put("530000","云南");
        map.put("500000","重庆");
        map.put("540000","西藏");
        // 返回集合
        return map ;
    }

}




